package com.young.sync.factory;

import com.young.canal.serializer.EventRecord;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @description：消费者组工厂
 * @author: yangyahui01
 * @date: 12/5/24 4:50 PM
 */
public class ConsumeListenerFactory {

    private static Map<String, ConcurrentKafkaListenerContainerFactory<String, EventRecord>> services = new ConcurrentHashMap<String, ConcurrentKafkaListenerContainerFactory<String, EventRecord>>();

    /**
     * 注册
     * @param consumeName 消费者组名称
     * @param consumeFactory
     */
    public static void register(String consumeName, ConcurrentKafkaListenerContainerFactory<String, EventRecord> consumeFactory) {
        services.put(consumeName, consumeFactory);
    }



    public static List<ConcurrentKafkaListenerContainerFactory<String, EventRecord>> getAllContainerListener() {
        return new ArrayList<>(services.values());
    }

    public static boolean isValidContainerListener(String consumeName) {
        return services.containsKey(consumeName);
    }

    /**
     *
     * @param consumeName 消费者组名称
     * @return
     */
    public static ConcurrentKafkaListenerContainerFactory<String, EventRecord> getContainerListenerByTopic(String consumeName) {
        return services.get(consumeName);
    }


}
